package com.bfxy.disruptor.common.waitStraty;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import com.bfxy.disruptor.common.balance.Sequence;
import com.bfxy.disruptor.common.balance.SequenceBarrier;
import com.bfxy.disruptor.common.exception.AlertException;
import com.bfxy.disruptor.common.exception.TimeoutException;

//有锁的方式
public class TimeoutBlockingWaitStrategy implements WaitStrategy {
	private final Lock lock = new ReentrantLock();// 重入锁
	// Condition用于多线程的唤醒和阻塞。类似于java的wait和notify，
	private final Condition processorNotifyCondition = lock.newCondition();
	private final long timeoutInNanos;

	public TimeoutBlockingWaitStrategy(final long timeout, final TimeUnit units) {
		timeoutInNanos = units.toNanos(timeout);
	}

	@Override
	public long waitFor(final long sequence, final Sequence cursorSequence, final Sequence dependentSequence,
			final SequenceBarrier barrier) throws AlertException, InterruptedException, TimeoutException {
		long nanos = timeoutInNanos;// 超时时间

		long availableSequence;
		if ((availableSequence = cursorSequence.get()) < sequence) {
			lock.lock();
			try {
				// 是否有可用的Sequence
				while ((availableSequence = cursorSequence.get()) < sequence) {
					barrier.checkAlert();
					nanos = processorNotifyCondition.awaitNanos(nanos);// 阻塞时候有超时时间的
					if (nanos <= 0) {
						throw TimeoutException.INSTANCE;
					}
				}
			} finally {
				lock.unlock();
			}
		}

		while ((availableSequence = dependentSequence.get()) < sequence) {
			barrier.checkAlert();
		}

		return availableSequence;
	}

	@Override
	public void signalAllWhenBlocking() {
		lock.lock();
		try {
			processorNotifyCondition.signalAll();
		} finally {
			lock.unlock();
		}
	}

}
